package com.hrt.datasender.datacollect.drive.weigeng;

import java.net.InetSocketAddress;
import java.util.Date;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

import com.btcode.common.MyLog;
import com.btcode.log.ILog;

public class WeiGengSocketClient {
    
    private final int millTrans = 1000;
    
    private final int maskCodeFF = 0xff;
    
    private ILog log;
    private ClientBootstrap bootstrap;
    private ChannelFuture connectFeature;
    private Date lastHeartBreakTime;

    private int heartBreakInterval;
    private String ip;
    private int port;
    private String clientId;// 连接标识，ip_端口
    private boolean isRestartting = false; // 是否正在重启中

    public WeiGengSocketClient(String ip, int port, int heartBreakInterval) {
        this.ip = ip;
        this.port = port;
        this.heartBreakInterval = heartBreakInterval;
        clientId = ip + "_" + port;
        log = MyLog.getInstance().getLogger(getClass().getName() + "_" + clientId);
        updateLastHeartBreakTime();

        /**
         * 启动消费队列任务
         */
        ReconnectJob executer = new ReconnectJob();
        executer.start();
    }

    public void startNetty() {
        NioClientSocketChannelFactory channelFactory = new NioClientSocketChannelFactory(
                Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
        bootstrap = new ClientBootstrap(channelFactory);

        ChannelPipelineFactory channelPipelineFactory = new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() throws Exception {
                ChannelPipeline channelPipeline = Channels.pipeline();
                channelPipeline.addFirst("MessageDecode", new WeiGengFrameDecode());
                channelPipeline.addLast("log", new MessageLog());
                channelPipeline.addLast("msgHandle", new WeiGengMsgHandle());
                channelPipeline.addLast("logResult", new MessageLog());
                return channelPipeline;
            }
        };
        bootstrap.setOption("child.reuseAddress", true);
        bootstrap.setPipelineFactory(channelPipelineFactory);
        connect();
    }

    public void connect() {
        if (isRestartting()) {
            return;
        }
        log.info("连接服务器……");

        setRestartting(true);
        connectFeature = bootstrap.connect(new InetSocketAddress(ip, port));
        connectFeature.addListener(new ChannelFutureListener() {

            @Override
            public void operationComplete(ChannelFuture arg0) throws Exception {
                setRestartting(false);
                updateLastHeartBreakTime();
                if (arg0.isSuccess()) {
                    log.info("连接成功," + ip + ":" + port);
                }
                else {
                    log.info("连接失败," + ip + ":" + port);
                }
            }
        });
    }

    /**
     * 判断连接是否可用
     * 
     * @return
     */
    public boolean isActivite() {
        Date now = new Date();
        long timeDiff = (now.getTime() - lastHeartBreakTime.getTime()) / millTrans;

        if (timeDiff > heartBreakInterval) {
            return false;
        }
        else {
            return true;
        }
    }

    public void updateLastHeartBreakTime() {
        this.lastHeartBreakTime = new Date();
    }

    public String getIp() {
        return ip;
    }

    public int getPort() {
        return port;
    }

    public String getClientId() {
        return clientId;
    }

    public boolean isRestartting() {
        log.info("获取isRestartting:" + isRestartting);
        return isRestartting;
    }

    private void setRestartting(boolean isRestartting) {
        log.info("设置isRestartting:" + isRestartting);
        this.isRestartting = isRestartting;
    }

    private class MessageLog extends SimpleChannelUpstreamHandler {

        @Override
        public void messageReceived(ChannelHandlerContext ctx, MessageEvent msgEvent) throws Exception {
            super.messageReceived(ctx, msgEvent);

            updateLastHeartBreakTime();

            Object result = msgEvent.getMessage();

            if (result instanceof byte[]) {
                byte[] data = (byte[]) msgEvent.getMessage();

                StringBuffer strBuf = new StringBuffer();

                for (byte b : data) {
                    strBuf.append(b + " ");
                }

                log.info("收到消息：" + strBuf);
            }
        }
    }

    public void writeMsg(byte[] msg) {

        if (!isActivite()) {
            log.info("发送信息失败，服务断开，重连中……");
            connect();
            return;
        }

        StringBuffer strBuf = new StringBuffer();

        for (byte b : msg) {
            int v = b & maskCodeFF;
            String vs = Integer.toHexString(v);
            vs = vs.length() > 1 ? vs : "0" + vs;
            strBuf.append(vs + " ");
        }

        ChannelBuffer buffer = ChannelBuffers.buffer(msg.length);
        buffer.writeBytes(msg);

        ChannelFuture feature = connectFeature.getChannel().write(buffer);
        feature.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture arg0) throws Exception {
                updateLastHeartBreakTime();
                if (!arg0.isSuccess()) {
                    if (arg0.getCause() != null) {
                        log.info("发送失败:" + arg0.getCause().getMessage() + "\r\n" + new String(msg));
                    }
                    else {
                        log.info("发送失败:" + new String(msg));
                    }
                }
                else {
                    log.info("发送成功:" + strBuf.toString());
                }
            }
        });
    }

    class ReconnectJob extends Thread {
        @Override
        public void run() {
            while (true) {
                if (!isActivite()) {
                    log.info("重连服务器……");
                    connect();
                }
                try {
                    Thread.sleep(millTrans);
                } 
                catch (Exception e) {
                    log.error("重连线程ReconnectJob出错", e);
                }
            }
        }
    }

}
